package com.streamax.s17.mqs.business.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import com.streamax.s17.cloud.common.model.ResponseCode;
import com.streamax.s17.mqs.api.common.constont.MsgDirection;
import com.streamax.s17.mqs.api.model.dto.MsgRetrieveDTO;
import com.streamax.s17.mqs.api.model.request.MsgRetrieveRQ;
import com.streamax.s17.mqs.business.common.MqsBusinessException;
import com.streamax.s17.mqs.business.common.constant.DevMsgField;
import com.streamax.s17.mqs.business.config.InfluxDBConfig;
import com.streamax.s17.mqs.business.dao.entity.DeviceMessage;
import com.streamax.s17.mqs.business.dao.util.InfluxDBConnect;
import com.streamax.s17.mqs.business.service.IMsgQueryService;
import com.streamax.s17.mqs.business.util.StoreUtil;
import com.streamax.s17.mqs.business.util.TimeUtil;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.BeanWrapperImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * @Description
 * @ClassName MsgRetrieveServiceImpl
 * @Author liwei
 * @Date 2019/8/13 18:01
 * @Version 2.0.2
 */
@Service
@Slf4j
public class MsgRetrieveServiceImpl implements IMsgQueryService {

    @Autowired
    InfluxDBConfig dbConfig;

    private InfluxDBConnect influxDB;

    /**
     * 插入
     * @param measurement 表
     * tags 标签
     * fields 字段
     */
    @Override
    public void persistenceMsg(List<DeviceMessage> msgs, String measurement) {
        {
            setUpDB();
            if (null == msgs || msgs.size() == 0) {
                return;
            }
            Map<String, Object> fields = new HashMap<>();
            Map<String, String> tags = new HashMap<>();

            for (DeviceMessage msg : msgs) {
                //消息类型
//                tags.put("TAG_CMD", msg.getCmd());
                //协议类型
//                tags.put("TAG_PROTOCOL", msg.getProtocolType());


                fields.put(DevMsgField.SERVICE_INSTANCE_ID, msg.getServiceInstanceId());
                fields.put(DevMsgField.CHANNEL, msg.getChannel());
                fields.put(DevMsgField.DEV_ID, msg.getDevId());
                fields.put(DevMsgField.SENDTO, msg.getSendTo());
                fields.put(DevMsgField.CMD, msg.getCmd());
                fields.put(DevMsgField.OCCUR_TIME, msg.getOccurTime());
                fields.put(DevMsgField.OCCUR_TIME_VAL, msg.getOccurTimeVal());
                fields.put(DevMsgField.PROTOCOL_TYPE, msg.getProtocolType());
                fields.put(DevMsgField.CONTENT, msg.getContent());

                influxDB.insert(measurement, tags, fields);
                log.debug("---[DB]写入数据库成功: [{}]", fields.toString());
            }
        }
    }


    @Override
    public List<MsgRetrieveDTO> search(MsgRetrieveRQ request) {
        String sql = buildSql(request);
        log.info("-----[DB]execute sql:{}", sql);
        if (null == influxDB){
            setUpDB();
        }
        QueryResult results = influxDB.query(sql);
        if (results.getResults() == null) {
            log.warn("---[DB]根据请求:{}, 查询不到记录.", request.toString());
            throw new MqsBusinessException(ResponseCode.NOT_FOUND, "根据请求:{}, 查询不到记录.");
        }
        List<MsgRetrieveDTO> lists = new ArrayList<>();
        for (QueryResult.Result result : results.getResults()) {
            List<QueryResult.Series> series = result.getSeries();
            if (null == series){
                continue;
            }
            for (QueryResult.Series serie : series) {
//				Map<String, String> tags = serie.getTags();
                List<List<Object>> values = serie.getValues();
                List<String> columns = serie.getColumns();
                lists.addAll(getQueryData(columns, values));
            }
        }
        if (CollectionUtil.isNotEmpty(lists)){
            StoreUtil.store(lists);
        }
        return lists;
    }

    /***整理列名、行数据***/
    private List<MsgRetrieveDTO> getQueryData(List<String> columns, List<List<Object>> values) {
        List<MsgRetrieveDTO> lists = new ArrayList<>();

        for (List<Object> list : values) {
            MsgRetrieveDTO dto = new MsgRetrieveDTO();
            BeanWrapperImpl bean = new BeanWrapperImpl(dto);
            boolean isN9M = false;
            for (int i = 0; i < list.size(); i++) {
                String propertyName = setColumns(columns.get(i));//字段名
                if ("occurTimeVal".equals(propertyName)){
                    continue;
                }
                Object value = list.get(i);//相应字段值
                if ("protocolType".equals(propertyName) && "n9m".equals(value)){
                    isN9M = true;
                }
                bean.setPropertyValue(propertyName, value);
            }

            if (isN9M) {
                lists.add(convertContFromHex(dto));
            }else{
                list.add(dto);
            }
        }

        return lists;
    }

    private MsgRetrieveDTO convertContFromHex(MsgRetrieveDTO originDTO){
        originDTO.setContent(fromHex(originDTO.getContent()));
        return originDTO;
    }

    /***转义字段***/
    private String setColumns(String column) {
        String[] cols = column.split("_");
        StringBuffer sb = new StringBuffer();
        for (int i = 0; i < cols.length; i++) {
            String col = cols[i];
//            String col = cols[i].toLowerCase();
            if (i != 0) {
                String start = col.substring(0, 1).toUpperCase();
                String end = col.substring(1).toLowerCase();
                col = start + end;
            }
            sb.append(col);
        }
        return sb.toString();
    }

    private String buildSql(MsgRetrieveRQ request){
        String tableName = dbConfig.getMeasurement();
        StringBuilder sb = new StringBuilder();
        sb.append("select * from ");
        sb.append(tableName);
        sb.append(buildWhereCondition(request));
        return sb.toString();
    }

    private String buildWhereCondition(MsgRetrieveRQ request){
        String beginTime = request.getBeginTime();
        String endTime = request.getEndTime();
        String devId = request.getDevId();
        MsgDirection direction = request.getDirection();
        List<String> whereList = new ArrayList<>();
        if (null != beginTime){
            whereList.add(" occurTimeVal>=" + TimeUtil.getCompleteNumFromStr(beginTime, 17));
        }
        if (null != endTime){
            whereList.add(" occurTimeVal<=" + TimeUtil.getCompleteNumFromStr(endTime, 17));
        }
        if (null != devId){
            whereList.add(" devId='" + devId);
        }
        if (null != direction && !direction.equals(MsgDirection.ALL)) {
            whereList.add(" sendTo='" + direction.name() + "'");
        }
        int whereSize = whereList.size();
        if (0 == whereSize){
            return "";
        }else if (1 == whereSize){
            return " where" + whereList.get(0);
        }else {
            StringBuffer sb = new StringBuffer();
            for (int i = 0; i < whereSize; i++) {
                if (i == 0){
                    sb.append(" where");
                }else{
                    sb.append(" and");
                }
                sb.append(whereList.get(i));
            }
            return sb.toString();
        }
    }

    /**
     * @Description: 根据配置设置数据库
     */
    public void setUpDB() {
        influxDB = new InfluxDBConnect(dbConfig);
        //连接数据库
        influxDB.influxDbBuild();
        //设置数据保存策略
        influxDB.createRetentionPolicy();
    }


    public static String fromHex(String s){
        if (s == null || s.equals("")) {
            return null;
        }
        s = s.replace(" ", "");
        byte[] baKeyword = new byte[s.length() / 2];
        for (int i = 0; i < baKeyword.length; i++) {
            try {
                baKeyword[i] = (byte) (0xff & Integer.parseInt(s.substring(i * 2, i * 2 + 2), 16));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        try {
            s = new String(baKeyword, "GBK");
            new String();
        } catch (Exception e1) {
            e1.printStackTrace();
        }
        String res = s.replaceAll("\\p{C}", "");
        int begin = res.indexOf("{");
        if (begin > 0){
            return res.substring(begin);
        }else {
            return "";
        }
    }


    private static int toDigit(char ch, int index) {
        int digit = Character.digit(ch, 16);
        if (digit == -1) {
            throw new RuntimeException("Illegal hexadecimal character " + ch + " at index " + index);
        }
        return digit;
    }

}
